// @Title business.go
// @Description cava smu服务主要业务逻辑
// @Author yezhibin 2021年5月1日01:22:46
// @Update

package business

import (
	"context"
	"encoding/json"
	"fmt"
	"math/rand"
	"net/http"
	"sort"
	"strconv"
	"strings"
	"time"

	"Projects/CavaCloudSys/cavasmusrc/assets"
	qfcfg "Projects/CavaCloudSys/cavasmusrc/com.qfcfg"
	qferror "Projects/CavaCloudSys/cavasmusrc/com.qferror"
	qfmail "Projects/CavaCloudSys/cavasmusrc/com.qfmail"
	util "Projects/CavaCloudSys/cavasmusrc/com.util"

	"go.uber.org/zap"

	"github.com/go-basic/uuid"
	"github.com/kataras/iris/v12"
	"github.com/pkg/errors"
	qflog "github.com/qf-tech/qftech-base/pkg/log"
	"nhooyr.io/websocket"
	"nhooyr.io/websocket/wsjson"
)

// Controller def
type Controller struct {
}

// Init def
func (control Controller) Init() {
	// 初始化消息接收缓存通道，size=200，并启动接收消息协程
	recvMsgChan = make(chan RecvRemoteMsg, 200)
	go control.recvDataLoop(recvMsgChan)

	// 初始化消息发送缓存通道，size=200，并启动发送消息协程
	sendMsgChan = make(chan SendRemoteMsg, 200)
	go control.sendDataLoop(sendMsgChan)

	teuInfoMap = make(map[string]*TeuInfo)

	clientInfoMap = make(map[string]*ClientInfo)

	registerClientInfoMap = make(map[string]*RegisterClientInfo)

	msgTypeMap = map[int]string{
		cavaClientRegisterReqMsg:    "cavaClientRegisterReqMsg",
		cavaClientRegisterRespMsg:   "cavaClientRegisterRespMsg",
		cavaTaskReqMsg:              "cavaTaskReqMsg",
		cavaTaskRespMsg:             "cavaTaskRespMsg",
		cavaTaskReportNotifyReqMsg:  "cavaTaskReportNotifyReqMsg",
		cavaTaskReportNotifyRespMsg: "cavaTaskReportNotifyRespMsg",
		cavaRegisterReqMsg:          "cavaRegisterReqMsg",
		cavaRegisterRespMsg:         "cavaRegisterRespMsg",
		cavaTaskStatusReqMsg:        "cavaTaskStatusReqMsg",
		cavaTaskStatusRespMsg:       "cavaTaskStatusRespMsg",
		cavaKeepAliveReqMsg:         "cavaKeepAliveReqMsg",
		cavaKeepAliveRespMsg:        "cavaKeepAliveRespMsg",
	}

	taskReqChanForLinux = make(chan TaskReqMsg, 100)
	taskReqChanForWin = make(chan TaskReqMsg, 100)

	// linux、windows任务分别由对应的分发协程和管道控制
	go control.disPatchTask(taskReqChanForLinux, "linux")
	go control.disPatchTask(taskReqChanForWin, "windows")

	go control.checkHeartBeatLoop()
}

func (control Controller) checkTraceID(r *http.Request) string {
	traceID := r.Header.Get("TRACE-ID")
	if traceID == "" {
		return control.genTraceID()
	}
	return traceID
}

func (control Controller) genTraceID() string {
	uid := uuid.New()
	tmpSlice := strings.Split(uid, "-")
	slicelen := len(tmpSlice)
	traceID := tmpSlice[slicelen-1]
	return traceID
}

// transToStr def
func (control Controller) transToStr(v interface{}) string {
	mJSON, err := json.Marshal(v)
	if err != nil {
		qflog.Sugare.Errorf("err:%v", err)
		return ""
	}
	msg := string(mJSON)
	return msg
}

// sendMsgToRemote def
func (control Controller) sendMsgToRemote(clientMsg SendRemoteMsg) {
	msg := control.transToStr(clientMsg.msg)
	dataMap := control.jsonToMap(msg)

	// 获取消息体中的traceID，获取不到则使用发送结构体里的traceID字段
	var traceID string
	if tmpID, ok := dataMap["trace_id"]; ok {
		traceID = fmt.Sprintf("%v", tmpID)
	}
	if traceID == "" {
		traceID = clientMsg.traceID
	}

	conn := clientMsg.conn
	err := wsjson.Write(clientMsg.ctx, conn, clientMsg.msg)
	if err != nil {
		var closeErr websocket.CloseError
		if errors.As(err, &closeErr) {
			qflog.Sugare.Warnf("TraceID=%s|conn already closed, client addr=%s", traceID, clientMsg.clientAddr)
		}
		qflog.Sugare.Errorf("err:%v", err)
	}
	qflog.Sugare.Infof("TraceID=%s|send data to client, data=%s, client addr=%s", traceID, msg, clientMsg.clientAddr)
}

func (control Controller) pushSendMsg(msgSt RecvRemoteMsg, sendMsg interface{}) {
	var sendMsgSt SendRemoteMsg
	sendMsgSt.conn = msgSt.conn
	sendMsgSt.clientAddr = msgSt.clientAddr
	sendMsgSt.msg = sendMsg
	sendMsgSt.ctx = msgSt.ctx
	sendMsgSt.traceID = msgSt.traceID
	sendMsgChan <- sendMsgSt
}

// sendDataLoop def
func (control Controller) sendDataLoop(ch chan SendRemoteMsg) {
	for msgSt := range ch {
		control.sendMsgToRemote(msgSt)
	}
}

func JSONResp(c iris.Context, status int, v interface{}) {
	log := qflog.GetLogger().GetCtx(c.Request().Context()).Sugar()
	c.StatusCode(status)

	resp := map[string]interface{}{
		"data": v,
	}

	_, err := c.JSON(resp)
	if err != nil {
		log.Errorf("Failed to encode: %v", err)
		return
	}

	retMsg, _ := json.Marshal(resp)
	log.Infof("response data: %s", string(retMsg))
}

// jsonToMap def
func (control Controller) jsonToMap(str string) map[string]interface{} {

	var tempMap map[string]interface{}
	err := json.Unmarshal([]byte(str), &tempMap)
	if err != nil {
		qflog.Sugare.Errorf("err:%v", err)
		return tempMap
	}

	return tempMap
}

func (control Controller) dispatchIdleTeu(sysType string) (*TeuInfo, error) {
	m := make(map[string]*TeuInfo)
	teuInfoMapMutex.Lock()
	defer teuInfoMapMutex.Unlock()
	for k, tmpTeuInfo := range teuInfoMap {
		if tmpTeuInfo.teuStatus == teuTaskStatusIdle && tmpTeuInfo.sysType == sysType {
			m[k] = tmpTeuInfo
		}
	}

	// 随机取出1个空闲teu
	teuCount := len(m)
	if teuCount > 0 {
		rand.Seed(time.Now().UnixNano())
		index := rand.Intn(100) % teuCount
		i := 0
		for _, tmpTeuInfo := range m {
			if index == i {
				return tmpTeuInfo, qferror.ErrOK
			}
			i++
		}
	}
	newTeuInfo := &TeuInfo{}
	return newTeuInfo, qferror.ErrNoIdleTeu
}

func (control Controller) getExistTeu(sysType string) (*TeuInfo, *qferror.Error) {
	teuInfoMapMutex.Lock()
	defer teuInfoMapMutex.Unlock()
	for _, tmpTeuInfo := range teuInfoMap {
		if tmpTeuInfo.sysType == sysType {
			return tmpTeuInfo, qferror.ErrOK
		}
	}
	newTeuInfo := &TeuInfo{}
	return newTeuInfo, qferror.ErrNoTeu
}

func (control Controller) disPatchTask(taskChan chan TaskReqMsg, sysType string) {
	for {
		// 如果能获取到空闲的teu，则等待请求任务从对应通道取出，无任务则阻塞，超时为30s。
		// 无空闲teu则继续尝试获取
		if idleTeuInfo, err := control.dispatchIdleTeu(sysType); err == qferror.ErrOK {

			// 超时机制
			timeout := make(chan bool, 1)
			go func() {
				time.Sleep(30 * time.Second)
				timeout <- true
			}()

			select {
			case task := <-taskChan:
				var msgSt RecvRemoteMsg
				msgSt.conn = idleTeuInfo.teuConn
				msgSt.clientAddr = idleTeuInfo.clientAddr
				msgSt.msg = nil
				msgSt.ctx = idleTeuInfo.ctx
				msgSt.traceID = task.TraceID
				control.updateTeuTaskStatus(idleTeuInfo.teuID, teuTaskStausTesting)
				control.pushSendMsg(msgSt, task)
				qflog.Sugare.Warnf("TraceID=%s|dispatch task to teu, teuID=%s, sysType=%s", msgSt.traceID, idleTeuInfo.teuID, task.SysType)
			case <-timeout:
				qflog.Sugare.Warnf("have no task over 30s, loop continue, sysType=%s", sysType)
			}
		}

		time.Sleep(200 * time.Millisecond)
	}
}

// checkHeartBeat def
func (control Controller) checkHeartBeat() {
	teuInfoMapMutex.Lock()
	defer teuInfoMapMutex.Unlock()

	for _, tmpTeuInfo := range teuInfoMap {
		nowTime := int32(time.Now().Unix())
		if nowTime-tmpTeuInfo.aliveTime >= 120 || tmpTeuInfo.aliveTime-nowTime >= 120 {
			qflog.Sugare.Warnf("teu have no heartbeat, connect will close, now:%d, aliveTime:%d, client addr:%s, teuID=%s.",
				nowTime, tmpTeuInfo.aliveTime, tmpTeuInfo.clientAddr, tmpTeuInfo.teuID)
			// _, cancel := context.WithTimeout(tmpTeuInfo.ctx, time.Second*1)
			// defer cancel()
			tmpTeuInfo.teuConn.Close(websocket.StatusNormalClosure, "")
		}
	}
}

// checkClientRegisterLoop def
func (control Controller) checkClientRegisterLoop(clientAddr string) {
	// 3 分钟循环
	i := 0
	var clienInfo *RegisterClientInfo
	var err *qferror.Error
	for i = 0; i < 9; i++ {
		if clienInfo, err = control.getRegisterClientConn(clientAddr); err == qferror.ErrOK {
			if clienInfo.isValid {
				break
			}
		}
		time.Sleep(20 * time.Second)
	}

	// 关闭客户端连接
	if i == 9 {
		qflog.Sugare.Warnf("client register timeout, client addr:%s, timeout=3min.", clientAddr)
		if err == qferror.ErrOK {
			qflog.Sugare.Warnf("next close client conn, client addr:%s, clientInfo:%v ", clientAddr, *clienInfo)
			clienInfo.clientConn.Close(websocket.StatusNormalClosure, "")
		}
	}
}

// checkHeartBeatLoop def
func (control Controller) checkHeartBeatLoop() {
	for {
		control.checkHeartBeat()
		time.Sleep(20 * time.Second)
	}
}

// doClientRegisterReqMsg def
func (control Controller) doClientRegisterReqMsg(msgSt RecvRemoteMsg) {
	msg := control.transToStr(msgSt.msg)
	var reqMsgSt ClientRegisterReqMsg
	_ = json.Unmarshal([]byte(msg), &reqMsgSt)

	// 待发送的消息写入对应channel
	respMsgSt := ClientRegisterRespMsg{
		MsgCmd:  cavaClientRegisterRespMsg,
		NetID:   "cava_smu",
		Code:    200,
		RetMsg:  "ok",
		TraceID: reqMsgSt.TraceID,
	}

	// CavaSvrProxy /authorization 鉴权
	server := qfcfg.ServerConfig.Server
	url := fmt.Sprintf("http://%s:%d/authorization", server.HttpIP, server.HttpPort)
	var requst = struct {
		Username  string `json:"username"`
		Password  string `json:"pwd"`
		Random    string `json:"random"`
		Timestamp string `json:"timestamp"`
	}{
		Username:  reqMsgSt.Username,
		Password:  reqMsgSt.Nonce,
		Random:    reqMsgSt.Random,
		Timestamp: reqMsgSt.Timestamp,
	}

	headers := map[string]string{
		"Content-Type": "application/json;charset=UTF-8",
	}

	v, err := util.HttpPostJson(msgSt.ctx, url, requst, headers, 10*time.Second)
	if err != nil {
		respMsgSt.RetMsg = err.Error()
		respMsgSt.Code = http.StatusBadRequest
		control.pushSendMsg(msgSt, respMsgSt)
		qflog.Sugare.Errorf("TraceID=%s|authorization err: %v, clientAddr: %s", msgSt.traceID, err, msgSt.clientAddr)
		return
	}

	qflog.Sugare.Infof("TraceID=%s|authorization resp body: %v, clientAddr: %s", msgSt.traceID, v, msgSt.clientAddr)
	type response struct {
		Data string `json:"data"`
	}
	respMsg := control.transToStr(v)
	var resp response
	_ = json.Unmarshal([]byte(respMsg), &resp)
	if resp.Data == "pass" {
		control.updateRegisterClientValid(msgSt.clientAddr, true)
		qflog.Sugare.Infof("TraceID=%s|client register ok, clientAddr: %s", msgSt.traceID, msgSt.clientAddr)
	}

	control.pushSendMsg(msgSt, respMsgSt)
}

type TaskInfo struct {
	TaskID     string
	TaskStatus string
	TaskName   string
	PassRate   string
	TaskReport string
	TaskLog    string
	TaskHtml   string
	TraceID    string
	Ctx        context.Context
}

func (control Controller) UpdateTaskInfo(req TaskInfo) error {
	server := qfcfg.ServerConfig.Server
	url := fmt.Sprintf("http://%s:%d/cava3/task", server.HttpIP, server.HttpPort)
	var requst = struct {
		UID        string `json:"uid"`
		TaskName   string `json:"taskName"`
		Status     string `json:"status"`
		PassRate   string `json:"passRate"`
		ReportPath string `json:"reportPath"`
		LogPath    string `json:"logPath"`
		HtmlPath   string `json:"htmlPath"`
	}{
		UID:        req.TaskID,
		TaskName:   req.TaskName,
		Status:     req.TaskStatus,
		PassRate:   req.PassRate,
		ReportPath: req.TaskReport,
		LogPath:    req.TaskLog,
		HtmlPath:   req.TaskHtml,
	}

	wsse := util.GenWsse(server.Username, server.Password, control.genTraceID())
	headers := map[string]string{
		"X-WSSE":       wsse,
		"Content-Type": "application/json;charset=UTF-8",
		"TRACE-ID":     req.TraceID,
	}

	v, err := util.HttpPostJson(req.Ctx, url, requst, headers, 10*time.Second)
	if err != nil {
		qflog.Sugare.Errorf("TraceID=%s|update task err: %v, uid: %s", req.TraceID, err)
		return errors.Wrap(err, "update task")
	}

	qflog.Sugare.Infof("TraceID=%s|update task resp body: %v", req.TraceID, v)

	type Err struct {
		Message string `json:"message"`
		Detail  string `json:"detail,omitempty"`
		Code    int    `json:"code,omitempty"`
	}

	type response struct {
		Data string `json:"data"`
	}
	respMsg := control.transToStr(v)
	var errResp Err
	var resp response
	err = json.Unmarshal([]byte(respMsg), &errResp)

	// 先按 error msg 解析，错误则表示有 data
	if err != nil || errResp.Message == "" {
		_ = json.Unmarshal([]byte(respMsg), &resp)
	} else {
		return errors.New(errResp.Message)
	}

	if strings.Contains(resp.Data, "succeed") {
		return nil
	}

	return errors.New("unknown err")
}

// doTaskReqMsg def
func (control Controller) doTaskReqMsg(msgSt RecvRemoteMsg) {
	// 待发送的消息写入对应channel
	var respMsgSt TaskRespMsg

	msg := control.transToStr(msgSt.msg)
	var reqMsgSt TaskReqMsg
	_ = json.Unmarshal([]byte(msg), &reqMsgSt)

	respMsgSt.MsgCmd = cavaTaskRespMsg
	respMsgSt.TaskID = reqMsgSt.TaskID
	respMsgSt.TaskName = reqMsgSt.TaskName
	respMsgSt.NetID = "cava_smu"
	respMsgSt.Code = 200
	respMsgSt.RetMsg = "ok"
	respMsgSt.TraceID = reqMsgSt.TraceID

	// client 注册鉴权判断
	if clienInfo, err := control.getRegisterClientConn(msgSt.clientAddr); err == qferror.ErrOK {
		if !clienInfo.isValid {
			respMsgSt.Code = qferror.ErrClientNeedAuth.Code()
			respMsgSt.RetMsg = qferror.ErrClientNeedAuth.Msg()
			qflog.Sugare.Errorf("TraceID=%s|client haven't authorized or authorize failed, clientAddr: %s", msgSt.traceID, msgSt.clientAddr)
			control.pushSendMsg(msgSt, respMsgSt)
			return
		}
	}

	// 不存在已注册的teu，则返回client相应错误提示
	if _, err := control.getExistTeu(reqMsgSt.SysType); err == qferror.ErrNoTeu {
		respMsgSt.Code = err.Code()
		respMsgSt.RetMsg = err.Msg()
		control.pushSendMsg(msgSt, respMsgSt)
		return
	}

	control.pushSendMsg(msgSt, respMsgSt)

	// 任务请求写入对应的分发通道
	// linux、windows任务分别由对应的分发协程和管道控制
	if strings.ToLower(reqMsgSt.SysType) == "linux" {
		taskReqChanForLinux <- reqMsgSt
	} else if strings.ToLower(reqMsgSt.SysType) == "windows" {
		taskReqChanForWin <- reqMsgSt
	}

	// 上报任务状态 testing
	go func() {
		req := TaskInfo{
			TaskID:     reqMsgSt.TaskID,
			TaskStatus: "testing",
			TaskName:   reqMsgSt.TaskName,
			TraceID:    reqMsgSt.TraceID,
			Ctx:        msgSt.ctx,
		}
		if err := control.UpdateTaskInfo(req); err != nil {
			qflog.Sugare.Errorf("TraceID=%s|update task info taskID: %s, status:%s, err: %v", req.TraceID, req.TaskID, req.TaskStatus, err)
		}
	}()

	newClientInfo := &ClientInfo{
		clientConn:    msgSt.conn,
		taskID:        reqMsgSt.TaskID,
		clientAddr:    msgSt.clientAddr,
		mailAddr:      reqMsgSt.MailAddr,
		ctx:           msgSt.ctx,
		blockPassMail: reqMsgSt.BlockPassMail,
	}
	control.updateClientConnection(reqMsgSt.TaskID, newClientInfo)
}

// doKeepAliveMsg def
func (control Controller) doKeepAliveMsg(msgSt RecvRemoteMsg) {
	msg := control.transToStr(msgSt.msg)
	var reqMsgSt KeepAliveReqMsg
	_ = json.Unmarshal([]byte(msg), &reqMsgSt)

	// 待发送的消息写入对应channel
	var respMsgSt KeepAliveRespMsg
	respMsgSt.MsgCmd = cavaKeepAliveRespMsg
	respMsgSt.NetID = "cava_smu"
	respMsgSt.Code = 200
	respMsgSt.RetMsg = "ok"
	respMsgSt.TraceID = reqMsgSt.TraceID

	control.pushSendMsg(msgSt, respMsgSt)
	control.updateTeuAliveTime(msgSt.clientAddr)
}

func (control Controller) updateTeuBaseInfo(ctx context.Context, teuID string, sysType string, clientAddr string) {
	teuInfoMapMutex.Lock()
	defer teuInfoMapMutex.Unlock()
	if tmpTeuInfo, ok := teuInfoMap[teuID]; ok {
		tmpTeuInfo.clientAddr = clientAddr
		tmpTeuInfo.ctx = ctx
		tmpTeuInfo.teuID = teuID
		tmpTeuInfo.sysType = sysType
		tmpTeuInfo.aliveTime = int32(time.Now().Unix())
		tmpTeuInfo.createdAt = tmpTeuInfo.aliveTime
		return
	}

	nowTime := int32(time.Now().Unix())
	newTeuInfo := TeuInfo{
		teuStatus:  teuTaskStatusIdle,
		teuConn:    nil,
		teuID:      teuID,
		clientAddr: clientAddr,
		sysType:    sysType,
		ctx:        ctx,
		aliveTime:  nowTime,
		createdAt:  nowTime,
	}
	teuInfoMap[teuID] = &newTeuInfo
}

func (control Controller) updateTeuTaskStatus(teuID string, status int) {
	teuInfoMapMutex.Lock()
	defer teuInfoMapMutex.Unlock()
	if tmpTeuInfo, ok := teuInfoMap[teuID]; ok {
		tmpTeuInfo.teuStatus = status
		return
	}

	newTeuInfo := TeuInfo{teuStatus: status, teuConn: nil, teuID: teuID, clientAddr: "", ctx: nil}
	teuInfoMap[teuID] = &newTeuInfo
}

func (control Controller) updateTeuConnection(teuID string, conn *websocket.Conn) {
	teuInfoMapMutex.Lock()
	defer teuInfoMapMutex.Unlock()
	if tmpTeuInfo, ok := teuInfoMap[teuID]; ok {
		tmpTeuInfo.teuConn = conn
		return
	}

	newTeuInfo := TeuInfo{teuStatus: teuTaskStatusIdle, teuConn: conn, teuID: teuID, clientAddr: "", ctx: nil}
	teuInfoMap[teuID] = &newTeuInfo
}

func (control Controller) updateTeuAliveTime(clientAddr string) {
	teuInfoMapMutex.Lock()
	defer teuInfoMapMutex.Unlock()

	for _, tmpTeuInfo := range teuInfoMap {
		if tmpTeuInfo.clientAddr == clientAddr {
			tmpTeuInfo.aliveTime = int32(time.Now().Unix())
			qflog.Sugare.Warnf("udpate heartbeat alive time:%d, teuID: %s, client addr:%s", tmpTeuInfo.aliveTime, tmpTeuInfo.teuID, tmpTeuInfo.clientAddr)
			break
		}
	}
}

func (control Controller) delTeuConnection(clientAddr string) {
	teuInfoMapMutex.Lock()
	defer teuInfoMapMutex.Unlock()
	teuID := ""
	for tmpTeuID, tmpTeuInfo := range teuInfoMap {
		if tmpTeuInfo.clientAddr == clientAddr {
			teuID = tmpTeuID
			break
		}

	}

	if teuID != "" {
		delete(teuInfoMap, teuID)
		qflog.Sugare.Infof("del form teuInfoMap, teuID: %s", teuID)
	}
}

func (control Controller) updateClientConnection(taskID string, clientInfo *ClientInfo) {
	clientInfoMapMutex.Lock()
	defer clientInfoMapMutex.Unlock()

	clientInfoMap[taskID] = clientInfo
}

func (control Controller) getClientConnection(taskID string) (*ClientInfo, *qferror.Error) {
	clientInfoMapMutex.Lock()
	defer clientInfoMapMutex.Unlock()
	if tmpCLientInfo, ok := clientInfoMap[taskID]; ok {
		return tmpCLientInfo, qferror.ErrOK
	}

	newClientInfo := ClientInfo{clientConn: nil, taskID: taskID, clientAddr: "", ctx: nil}
	clientInfoMap[taskID] = &newClientInfo
	return &newClientInfo, qferror.ErrNoClientInfo
}

func (control Controller) delClientConnection(taskID string) {
	clientInfoMapMutex.Lock()
	defer clientInfoMapMutex.Unlock()
	if _, ok := clientInfoMap[taskID]; ok {
		delete(clientInfoMap, taskID)
		qflog.Sugare.Infof("del form clientInfoMap, taskID: %s", taskID)
	}
}

func (control Controller) updateRegisterClientConn(clientAddr string, clientInfo *RegisterClientInfo) {
	registerClientInfoMapMutex.Lock()
	defer registerClientInfoMapMutex.Unlock()

	registerClientInfoMap[clientAddr] = clientInfo
}

func (control Controller) updateRegisterClientValid(clientAddr string, valid bool) {
	registerClientInfoMapMutex.Lock()
	defer registerClientInfoMapMutex.Unlock()
	if tmpClientInfo, ok := registerClientInfoMap[clientAddr]; ok {
		tmpClientInfo.isValid = valid
		return
	}
}

func (control Controller) getRegisterClientConn(clientAddr string) (*RegisterClientInfo, *qferror.Error) {
	registerClientInfoMapMutex.Lock()
	defer registerClientInfoMapMutex.Unlock()
	if tmpCLientInfo, ok := registerClientInfoMap[clientAddr]; ok {
		return tmpCLientInfo, qferror.ErrOK
	}

	return nil, qferror.ErrNoClientInfo
}

func (control Controller) delRegisterClientConn(clientAddr string) {
	registerClientInfoMapMutex.Lock()
	defer registerClientInfoMapMutex.Unlock()
	if _, ok := registerClientInfoMap[clientAddr]; ok {
		delete(registerClientInfoMap, clientAddr)
		qflog.Sugare.Infof("del from registerClientInfoMap, clientAddr: %s", clientAddr)
	}
}

func (control Controller) doRegisterReqMsg(msgSt RecvRemoteMsg) {
	msg := control.transToStr(msgSt.msg)
	var reqMsgSt RegisterReqMsg
	_ = json.Unmarshal([]byte(msg), &reqMsgSt)

	// 待发送的消息写入对应channel
	var respMsgSt RegisterRespMsg
	respMsgSt.MsgCmd = cavaRegisterRespMsg
	respMsgSt.NetID = "cava_smu"
	respMsgSt.Code = 200
	respMsgSt.RetMsg = "ok"
	respMsgSt.TraceID = reqMsgSt.TraceID

	control.updateTeuBaseInfo(msgSt.ctx, reqMsgSt.TeuID, reqMsgSt.SysType, msgSt.clientAddr)
	control.updateTeuTaskStatus(reqMsgSt.TeuID, teuTaskStatusIdle)
	control.updateTeuConnection(reqMsgSt.TeuID, msgSt.conn)

	// teu 注册无需鉴权
	control.updateRegisterClientValid(msgSt.clientAddr, true)
	qflog.Sugare.Infof("TraceID=%s|teu register ok, teuID: %s, clientAddr: %s", msgSt.traceID, reqMsgSt.TeuID, msgSt.clientAddr)

	control.pushSendMsg(msgSt, respMsgSt)
}

func (control Controller) sendReportMail(reqMsgSt TaskReportReqMsg, clienInfo *ClientInfo) error {
	// 发送邮件
	// 定义收件人
	var mailTo []string
	if strings.Contains(clienInfo.mailAddr, ";") {
		mailTo = strings.Split(clienInfo.mailAddr, ";")
	} else {
		mailTo = []string{
			clienInfo.mailAddr,
		}
	}
	// 邮件主题
	subject := fmt.Sprintf("【cava任务报告】%s", reqMsgSt.TaskName)
	// 邮件正文
	templatePath := "templates/mail/report/report.html"
	body, err := qfmail.MakeBodyFromTemplate(
		assets.MailTemplates,
		templatePath,
		map[string]interface{}{
			"taskName":     reqMsgSt.TaskName,
			"taskPassRate": fmt.Sprintf("%s, %s", reqMsgSt.PassRate, reqMsgSt.PassSummary),
			"taskID":       reqMsgSt.TaskID,
			"taskReport": func() string {
				if reqMsgSt.TaskHtml != "" {
					return reqMsgSt.TaskHtml
				}
				return reqMsgSt.TaskReport
			}(),
			"taskLog": reqMsgSt.TaskLog,
		},
	)
	if err != nil {
		return errors.Wrap(err, "make body")
	}

	noteSender := "qftech"
	err = qfmail.SendMail(mailTo, subject, string(body), noteSender)
	if err != nil {
		return err
	}
	return nil
}

func (control Controller) doReportReqMsg(msgSt RecvRemoteMsg) {
	msg := control.transToStr(msgSt.msg)
	var reqMsgSt TaskReportReqMsg
	_ = json.Unmarshal([]byte(msg), &reqMsgSt)

	// 待发送的消息写入对应channel
	var respMsgSt TaskReportRespMsg
	respMsgSt.MsgCmd = cavaTaskReportNotifyRespMsg
	respMsgSt.NetID = "cava_smu"
	respMsgSt.Code = 200
	respMsgSt.RetMsg = "ok"
	respMsgSt.TraceID = reqMsgSt.TraceID

	control.updateTeuTaskStatus(reqMsgSt.TeuID, teuTaskStatusIdle)

	control.pushSendMsg(msgSt, respMsgSt)

	// 转发结果给客户端
	if tmpClientInfo, ret := control.getClientConnection(reqMsgSt.TaskID); ret == qferror.ErrOK {
		// 上报任务状态 completed
		go func() {
			req := TaskInfo{
				TaskID:     reqMsgSt.TaskID,
				TaskStatus: "completed",
				TaskName:   reqMsgSt.TaskName,
				TraceID:    reqMsgSt.TraceID,
				PassRate:   fmt.Sprintf("%s, %s", reqMsgSt.PassRate, reqMsgSt.PassSummary),
				TaskReport: reqMsgSt.TaskReport,
				TaskLog:    reqMsgSt.TaskLog,
				TaskHtml:   reqMsgSt.TaskHtml,
				Ctx:        msgSt.ctx,
			}
			if err := control.UpdateTaskInfo(req); err != nil {
				qflog.Sugare.Errorf("TraceID=%s|update task info taskID: %s, status:%s, err: %v", req.TraceID, req.TaskID, req.TaskStatus, err)
			}
		}()

		msgSt.conn = tmpClientInfo.clientConn
		msgSt.clientAddr = tmpClientInfo.clientAddr
		msgSt.ctx = tmpClientInfo.ctx
		control.pushSendMsg(msgSt, reqMsgSt)

		// 发送邮件
		if tmpClientInfo.mailAddr != "" {
			// 测试通过率 100% 且 设置了全通过不发送邮件选项，则不发送邮件通知
			if reqMsgSt.PassRate == "100.00%" && tmpClientInfo.blockPassMail {
				qflog.Sugare.Warnf("TraceID=%s|TaskID=%s all passed, not send mail", respMsgSt.TraceID, reqMsgSt.TaskID)
			} else {
				var err error
				for i := 0; i < 5; i++ {
					err = control.sendReportMail(reqMsgSt, tmpClientInfo)
					if err == nil {
						break
					}
					time.Sleep(time.Second * 3)
				}
				if err != nil {
					qflog.Sugare.Errorf("TraceID=%s|send mail failed, err: %v", respMsgSt.TraceID, err)
				}
			}
		}
		control.delClientConnection(reqMsgSt.TaskID)
		qflog.Sugare.Infof("TraceID=%s|Trans teu[teu_id: %s] report to client address: %s, data: %s", respMsgSt.TraceID, reqMsgSt.TeuID, tmpClientInfo.clientAddr, msg)
	}
}

// processData def
func (control Controller) processData(recvMsg RecvRemoteMsg) {
	msg := control.transToStr(recvMsg.msg)
	dataMap := control.jsonToMap(msg)

	// 获取消息体中的traceID，获取不到则临时生成1个
	var traceID string
	if tmpID, ok := dataMap["trace_id"]; ok {
		traceID = fmt.Sprintf("%v", tmpID)
	}
	if traceID == "" {
		traceID = control.genTraceID()
	}

	if cmd, ok1 := dataMap["msg_cmd"]; ok1 {
		// interface{} 间接转int
		cmdStr := fmt.Sprintf("%v", cmd)
		cmdVal, _ := strconv.Atoi(cmdStr)
		qflog.Sugare.Infof("TraceID=%s|processData, cmd=%d, type=%s, recv data from client, data=%s, client addr=%s", traceID,
			cmdVal, msgTypeMap[cmdVal], msg, recvMsg.clientAddr)

		switch cmdVal {
		case cavaClientRegisterReqMsg:
			go control.doClientRegisterReqMsg(recvMsg)
		case cavaTaskReqMsg:
			go control.doTaskReqMsg(recvMsg)
		case cavaKeepAliveReqMsg:
			go control.doKeepAliveMsg(recvMsg)
		case cavaRegisterReqMsg:
			go control.doRegisterReqMsg(recvMsg)
		case cavaTaskReportNotifyReqMsg:
			go control.doReportReqMsg(recvMsg)

		}
	}
}

// recvDataLoop def
func (control Controller) recvDataLoop(ch chan RecvRemoteMsg) {
	for msgSt := range ch {
		control.processData(msgSt)
	}
}

// readData def
func (control Controller) readDataLoop(req *http.Request, conn *websocket.Conn, ch chan bool) {
	for {
		var v interface{}
		ctx := req.Context()
		err := wsjson.Read(ctx, conn, &v)
		if err != nil {
			var closeErr websocket.CloseError
			if errors.As(err, &closeErr) {
				qflog.Sugare.Warnf("conn already closed, client addr=%s", req.RemoteAddr)
			}
			ch <- true
			qflog.Sugare.Errorf("err:%v", err)
			break
		}
		msg := control.transToStr(v)
		dataMap := control.jsonToMap(msg)
		// 获取消息体中的traceID，获取不到则临时生成1个
		var traceID string
		if tmpID, ok := dataMap["trace_id"]; ok {
			traceID = fmt.Sprintf("%v", tmpID)
		}
		if traceID == "" {
			traceID = control.genTraceID()
		}

		newCtx, _ := qflog.GetLogger().AddCtx(ctx, zap.Any("TraceID", traceID))
		// 接收到的消息写入对应channel
		var msgSt RecvRemoteMsg
		msgSt.conn = conn
		msgSt.clientAddr = req.RemoteAddr
		msgSt.msg = v
		msgSt.ctx = newCtx
		msgSt.traceID = traceID
		recvMsgChan <- msgSt
		// qflog.Sugare.Infof("TraceID=%s|recv data from client, data=%s, client addr=%s", traceID, data, req.RemoteAddr)
	}
}

// WsHandle def
func (control Controller) WsHandle(w http.ResponseWriter, req *http.Request) {
	traceID := control.checkTraceID(req)
	var opts websocket.AcceptOptions
	opts.InsecureSkipVerify = true
	conn, err := websocket.Accept(w, req, &opts)
	if err != nil {
		qflog.Sugare.Errorf("err:%v", err)
		return
	}
	defer conn.Close(websocket.StatusInternalError, "内部出错")

	// ctx, cancel := context.WithTimeout(req.Context(), time.Second*10)
	ctx := req.Context()
	// defer cancel()

	qflog.Sugare.Infof("TraceID=%s|client connect succeed, client addr: %s", traceID, req.RemoteAddr)
	// err = wsjson.Write(ctx, conn, JSONResult{Status: 200, Data: "hello websocket client", ClientAddr: req.RemoteAddr})
	// if err != nil {
	// 	qflog.Sugare.Errorf("err:%v", err)
	// 	return
	// }

	// 创建连接关闭通道
	closeCh := make(chan bool, 1)

	// 开启读数据协程
	go control.readDataLoop(req, conn, closeCh)

	// 待发送的消息写入对应channel
	var msgSt SendRemoteMsg
	msgSt.conn = conn
	msgSt.clientAddr = req.RemoteAddr
	msgSt.msg = JSONResult{Status: 200, Data: "hello websocket client", ClientAddr: req.RemoteAddr}
	msgSt.ctx = ctx
	msgSt.traceID = traceID
	sendMsgChan <- msgSt

	// 更新客户端信息，包含teu和client
	newRegisterClientInfo := &RegisterClientInfo{
		clientConn: msgSt.conn,
		clientAddr: msgSt.clientAddr,
		ctx:        msgSt.ctx,
	}
	control.updateRegisterClientConn(req.RemoteAddr, newRegisterClientInfo)

	go control.checkClientRegisterLoop(req.RemoteAddr)

	// conn.Close(websocket.StatusNormalClosure, "")
	flag := <-closeCh
	control.delTeuConnection(req.RemoteAddr)
	control.delRegisterClientConn(req.RemoteAddr)
	qflog.Sugare.Warnf("TraceID=%s|exit go routinue: %v, client addr:%s", traceID, flag, req.RemoteAddr)
}

// CheckTraceID def
func (control Controller) CheckTraceID(c iris.Context) {
	traceID := c.GetHeader("TRACE-ID")
	if traceID == "" {
		uid := uuid.New()
		tmpSlice := strings.Split(uid, "-")
		slicelen := len(tmpSlice)
		traceID = tmpSlice[slicelen-1]
		c.Request().Header.Set("TRACE-ID", traceID)
		qflog.Sugare.Warnf("Request generate trace id:%s, uri:%s method:%s", traceID, c.Request().RequestURI, c.Method())
	}
	c.Header("TRACE-ID", traceID)
}

// GetTeuInfo def
func (control Controller) GetTeuInfo(ctx iris.Context) {
	type customTime struct {
		Unix  int32          `json:"unix"`
		Local util.LocalTime `json:"local"`
	}
	type teuInfo struct {
		Status     string         `json:"status"`
		UID        string         `json:"uid"`
		ClientAddr string         `json:"clientAddr"`
		SysType    string         `json:"sysType"`
		CreatedAt  util.LocalTime `json:"createdAt"`
		UpdatedAt  customTime     `json:"updatedAt"`
	}

	teuList := make([]*teuInfo, 0, 1)
	status := []string{"idle", "testing"}
	{
		teuInfoMapMutex.Lock()
		defer teuInfoMapMutex.Unlock()

		s := make([]*TeuInfo, 0, len(teuInfoMap))
		for _, e := range teuInfoMap {
			s = append(s, e)
		}
		// 降序排列
		sort.Slice(s, func(i, j int) bool {
			return s[i].createdAt > s[j].createdAt
		})

		for _, e := range s {
			teuList = append(teuList, &teuInfo{
				Status:     status[e.teuStatus],
				UID:        e.teuID,
				ClientAddr: e.clientAddr,
				SysType:    e.sysType,
				CreatedAt: func() util.LocalTime {
					createdAt := time.Unix(int64(e.createdAt), 0)

					return util.LocalTime(createdAt)
				}(),
				UpdatedAt: customTime{
					Unix: e.aliveTime,
					Local: func() util.LocalTime {
						aliveTime := time.Unix(int64(e.aliveTime), 0)
						return util.LocalTime(aliveTime)
					}(),
				},
			})
		}
	}

	var resp = struct {
		Total   int32      `json:"total"`
		TeuInfo []*teuInfo `json:"teuInfo"`
	}{
		Total:   int32(len(teuList)),
		TeuInfo: teuList,
	}

	JSONResp(ctx, http.StatusOK, resp)
}
